package com.atguigu.edu.realtime.app.dwd.db;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * Author: Felix
 * Date: 2022/5/10
 * Desc: 交易域加购事务事实表
 */
public class DwdTradeCartAdd {
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 设定 Table 中的时区为本地时区
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 2.检查点相关设置(略)


        // TODO 3. 从Kafka 读取业务数据，封装为Flink SQL表
        tableEnv.executeSql("" +
                "create table topic_db(" +
                "`database` string," +
                "`table` string," +
                "`type` string," +
                "`data` map<string, string>," +
                "`old` map<string, string>," +
                "`ts` string," +
                "`proc_time` as PROCTIME()" +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_cart_add"));

        // TODO 4. 从业务表中过滤出购物车表数据
       Table cartAdd = tableEnv.sqlQuery(
        "select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['course_id'] course_id," +
                "data['course_name'] course_name," +
                "data['cart_price'] cart_price,"+
                 "ts," +
                "proc_time" +
                " from `topic_db` " +
                " where `table` = 'cart_info'" +
                " and `type` = 'insert'" );

        tableEnv.createTemporaryView("cart_add", cartAdd);

        Table resultTable = tableEnv.sqlQuery("select " +
                        "cadd.id," +
                        "user_id," +
                        "course_id," +
                        "course_name," +
                        "cart_price,"+
                "ts" +
                        " from cart_add cadd " );
        tableEnv.createTemporaryView("result_table", resultTable);



        // TODO 7. 建立 Upsert-Kafka dwd_trade_cart_add 表
        tableEnv.executeSql("" +
                "create table dwd_trade_cart_add(" +
                "id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "cart_price string," +
                "ts string," +
                "primary key(id) not enforced" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cart_add"));

        // TODO 8. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_cart_add select * from result_table");
    }

}
